package com.example.mq.mqserver.datacenter;

import com.example.mq.common.exception.MqException;
import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/*
 * 使用这个类来统一管理内存中的所有数据.
 * 该类后续提供的一些方法, 可能会在多线程环境下被使用. 因此要注意线程安全问题.
 */
@Slf4j
public class MemoryDataCenter {
    // 管理所有的交换机
    private final ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();

    // 管理所有的消息队列
    private final ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();

    // 管理交换机与队列的绑定
    // 第一个 key 是 exchangeName, 第二个 key 是 queueName
    private final ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();

    // 管理所有的消息
    // key 是 messageId, value 是 Message 对象
    private final ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();

    // 管理队列下的所有消息
    // key 是 queueName, value 是一个 Message 的链表
    private final ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();

    // 管理未确认的消息
    // 第一个 key 是 queueName, 第二个 key 是 messageId
    private final ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    // 插入一个交换机到内存
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        log.info("[MemoryDataCenter] 新交换机添加成功! exchangeName={}", exchange.getName());
    }

    // 根据交换机名字获取交换机
    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    // 删除内存中的一个交换机
    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        log.info("[MemoryDataCenter] 交换机删除成功! exchangeName={}", exchangeName);
    }

    // 插入一个队列到内存
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        log.info("[MemoryDataCenter] 新队列添加成功! queueName={}", queue.getName());
    }

    // 根据队列名字获取队列
    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    // 删除内存中的一个队列
    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        log.info("[MemoryDataCenter] 队列删除成功! queueName={}", queueName);
    }

    // 插入一个绑定到内存
    public void insertBinding(Binding binding) throws MqException {
        // 先查一下交换机下有没有初始化 Map，有就获取，没有就初始化一个
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());

        // 同步块，确保绑定操作的线程安全性
        synchronized (bindingMap) {
            // 再查一下是否已经存在绑定（重复绑定）
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter] 绑定已经存在! exchangeName=" + binding.getExchangeName() + ", queueName=" + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
        }
        log.info("[MemoryDataCenter] 新绑定添加成功! exchangeName={}, queueName={}",
                binding.getExchangeName(), binding.getQueueName());
    }

    // 获取绑定, 写两个版本:

    // 1. 根据 exchangeName 和 queueName 确定唯一一个 Binding
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }

    // 2. 根据 exchangeName 获取到所有的 Binding
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    // 删除内存中的一个绑定
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            // 该交换机没有绑定任何队列. 报错.
            throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName=" + binding.getExchangeName()
                    + ", queueName=" + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        log.info("[MemoryDataCenter] 绑定删除成功! exchangeName={}, queueName={}",
                binding.getExchangeName(), binding.getQueueName());
    }

    // 添加消息到内存
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        log.info("[MemoryDataCenter] 新消息添加成功! messageId={}", message.getMessageId());
    }

    // 根据消息ID查询消息
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    // 根据消息ID删除消息
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        log.info("[MemoryDataCenter] 消息被移除! messageId={}", messageId);
    }

    // 发送消息到指定队列
    public void sendMessage(MSGQueue queue, Message message) {
        // 把消息放到对应的队列数据结构中.
        // 先根据队列的名字, 找到该队列对应的消息链表.
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
        // 再把数据加到 messages 里面
        synchronized (messages) {
            messages.add(message);
        }
        // 在这里把该消息也往消息中心中插入一下.
        addMessage(message);
        log.info("[MemoryDataCenter] 消息被投递到队列中! messageId={}", message.getMessageId());
    }

    // 从队列中取消息
    public Message pollMessage(String queueName) {
        // 根据队列名, 查找一下, 对应的队列的消息链表.
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return null;
        }
        synchronized (messages) {
            // 如果没找到, 说明队列中没有任何消息.
            if (messages.isEmpty()) {
                return null;
            }
            // 链表中有元素, 就进行头删.
            Message currentMessage = messages.removeFirst();
            log.info("[MemoryDataCenter] 消息从队列中取出! messageId={}", currentMessage.getMessageId());
            return currentMessage;
        }
    }

    // 获取指定队列中消息的个数
    public int getMessageCount(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            // 队列中没有消息
            return 0;
        }
        synchronized (messages) {
            return messages.size();
        }
    }

    // 添加未确认的消息
    public void addMessageWaitAck(String queueName, Message message) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.computeIfAbsent(queueName, k -> new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(), message);
        log.info("[MemoryDataCenter] 消息进入待确认队列! messageId={}", message.getMessageId());
    }

    // 删除未确认的消息(消息已经确认了)
    public void removeMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) {
            return;
        }
        messageHashMap.remove(messageId);
        log.info("[MemoryDataCenter] 消息从待确认队列删除! messageId={}", messageId);
    }

    // 获取指定的未确认的消息
    public Message getMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) {
            return null;
        }
        return messageHashMap.get(messageId);
    }

    // 从硬盘中恢复数据到内存
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {

        // 0. 清空之前的所有数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();

        // 1. 恢复所有的交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for (Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(), exchange);
        }

        // 2. 恢复所有的队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for (MSGQueue queue : queues) {
            queueMap.put(queue.getName(), queue);
        }

        // 3. 恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(), binding);
        }

        // 4. 恢复所有的消息数据
        for (MSGQueue queue : queues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(), messages);
            for (Message message : messages) {
                messageMap.put(message.getMessageId(), message);
            }
        }

        // 未确认的消息不需要从硬盘恢复，因为重启后这些消息将被视为未被取走的消息
    }
}
